package com.doyoo.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.doyoo.config.JobProperties;
import com.doyoo.entity.JobInfo;
import com.doyoo.entity.JobStep;
import com.doyoo.entity.JobStepTemplate;
import com.doyoo.enums.State;
import com.doyoo.enums.Type;
import com.doyoo.mapper.JobInfoMapper;
import com.doyoo.service.JobInfoService;
import com.doyoo.service.JobStepService;
import com.doyoo.service.JobStepTemplateService;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Maps;
import io.kubernetes.client.openapi.ApiException;
import io.kubernetes.client.openapi.apis.BatchV1Api;
import io.kubernetes.client.openapi.apis.BatchV1beta1Api;
import io.kubernetes.client.openapi.models.*;
import io.kubernetes.client.util.generic.GenericKubernetesApi;
import io.kubernetes.client.util.generic.options.DeleteOptions;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Service
public class JobInfoServiceImpl extends ServiceImpl<JobInfoMapper, JobInfo> implements JobInfoService {

    @Autowired
    private ObjectMapper objectMapper;

    @Autowired
    private JobStepService jobStepService;

    @Autowired
    private JobStepTemplateService jobStepTemplateService;

    @Autowired
    private BatchV1beta1Api batchV1beta1Api;

    @Autowired
    private BatchV1Api batchV1Api;

    @Autowired
    private GenericKubernetesApi<V1beta1CronJob, V1beta1CronJobList> cronJobGenericApi;

    @Autowired
    private DeleteOptions deleteOptions;

    @Autowired
    private JobProperties jobProperties;

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void addOne(JobInfo jobInfo) {
        // 插入数据
        this.save(jobInfo);
        jobInfo.getStepList().forEach(e -> e.setJobId(jobInfo.getId()));
        jobStepService.saveBatch(jobInfo.getStepList());
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void updateOne(JobInfo jobInfo) {

        LambdaQueryWrapper<JobInfo> wrapper = Wrappers.<JobInfo>lambdaQuery()
                .eq(JobInfo::getId, jobInfo.getId())
                .last("for update");
        String oriName = this.getOne(wrapper).getName();
        if (!StringUtils.equals(oriName, jobInfo.getName())) {
            throw new RuntimeException("作业名称不能修改");
        }
        // 更新数据
        this.updateById(jobInfo);
        jobStepService.remove(Wrappers.<JobStep>lambdaQuery().eq(JobStep::getJobId, jobInfo.getId()));
        jobStepService.saveBatch(jobInfo.getStepList());
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void enableOne(Integer id) throws JsonProcessingException, ApiException {
        updateState(id, State.ENABLED);

        JobInfo jobInfo = this.getById(id);
        if (jobInfo == null) {
            throw new RuntimeException("未找到作业");
        }

        V1beta1CronJob v1beta1CronJob = new V1beta1CronJobBuilder()
                .withNewMetadata().withName(jobInfo.getName()).endMetadata()
                .withNewSpec()
                    .withSchedule(jobInfo.getSchedule())
                    .withConcurrencyPolicy("Forbid")//Allow/Forbid/Replace
                    .withFailedJobsHistoryLimit(jobInfo.getFailedHistoryLimit())
                    .withSuccessfulJobsHistoryLimit(jobInfo.getSuccessfulHistoryLimit())
                    .withStartingDeadlineSeconds(15L)//默认
                    .withNewSuspend(false)

                    .withNewJobTemplate()
                        .withNewSpec()
                            .withNewTemplate()
                                .withNewMetadata().withAnnotations(Collections.singletonMap("sidecar.istio.io/inject", "false")).endMetadata()
                                .withNewSpec()
                                    .withContainers(
                                            new V1ContainerBuilder()
                                                    .withName("hello")
                                                    .withImage(jobProperties.getImages().get(jobInfo.getType().name()))
                                                    //.withArgs(Arrays.asList("/bin/sh", "-c", "date; echo Hello from the Kubernetes cluster"))
                                                    .withArgs(this.getArgs(jobInfo.getType(), id))
                                                    .withImagePullPolicy("IfNotPresent")
                                                    .build()
                                    )
                                    .withRestartPolicy("OnFailure")
                                    .withNewDnsPolicy("ClusterFirst")
                                .endSpec()
                            .endTemplate()
                        .endSpec()
                    .endJobTemplate()
                .endSpec()
            .build();
        batchV1beta1Api.createNamespacedCronJob(jobInfo.getEnvs(), v1beta1CronJob, null, null, null);
    }


    private void createJob(Integer id) throws JsonProcessingException, ApiException {
        JobInfo jobInfo = this.getById(id);
        if (jobInfo == null) {
            throw new RuntimeException("未找到作业");
        }
        //batchV1Api
        V1Job job = new V1JobBuilder()
                .withNewMetadata().withName(jobInfo.getName()).withLabels(Collections.singletonMap("job", jobInfo.getName())).endMetadata()
                .withNewSpec()
                    .withCompletions(1)  //成功运行Pods的次数。默认值: 1
                    .withParallelism(1)  //并发运行Pods的数量。默认值: 1
                    .withBackoffLimit(0) //失败重试次数。默认值：6次
                    .withActiveDeadlineSeconds(0L) //Pod多少秒就停止，Pod运行的超时时间(0表示运行直到正常结束)
                    .withNewTemplate()
                        .withNewSpec()
                            .withContainers(
                                    new V1ContainerBuilder()
                                            .withName("hello")
                                            .withImage(jobProperties.getImages().get(jobInfo.getType().name()))
                                            //.withArgs(Arrays.asList("/bin/sh", "-c", "date; echo Hello from the Kubernetes cluster"))
                                            // 注意command 和args可以合并
                                            //.withCommand(Arrays.asList("/bin/sh", "-c"))
                                            .withArgs(this.getArgs(jobInfo.getType(), id))
                                            .withImagePullPolicy("IfNotPresent")
                                            .build()
                            )
                            .withRestartPolicy("OnFailure")
                            .withNewDnsPolicy("ClusterFirst")
                        .endSpec()
                    .endTemplate()
                .endSpec()
            .build();
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void disableOne(Integer id) throws ApiException {
        updateState(id, State.DISABLED);
        JobInfo jobInfo = this.getById(id);
        if (jobInfo == null) {
            throw new RuntimeException("未找到作业");
        }
        cronJobGenericApi.delete(jobInfo.getEnvs(), jobInfo.getName(), deleteOptions).throwsApiException().getObject();
    }

    private void updateState(Integer id, State state) {
        JobInfo jobInfo = this.getOne(Wrappers.<JobInfo>lambdaQuery()
                .eq(JobInfo::getId, id)
                .last("for update"));
        if (jobInfo == null) {
            throw new RuntimeException("未找到作业");
        }
        if (state.getValue().equals(jobInfo.getState())) {
            throw new RuntimeException(String.format("该作业已经是[%s]状态", state.getName()));
        }

        JobInfo record = new JobInfo();
        record.setId(jobInfo.getId());
        record.setState(state);
        this.updateById(record);
    }

    private List<String> getArgs(Type type, Integer id) throws JsonProcessingException {
        List<String> args = new ArrayList<>();
        if (type == Type.SHELL) {
            args.add("/bin/sh");
            args.add("-c");
        } else if (type == Type.PYTHON) {
            args.add("python");
            args.add("-c");
        }

        // 得到JobStep
        List<JobStep> jobStepList = jobStepService.list(Wrappers.<JobStep>lambdaQuery()
                .eq(JobStep::getJobId, id)
                .orderByAsc(JobStep::getId));

        // 得到相应的JobStepTemplate
        List<Integer> templateIdList = jobStepList.stream()
                .map(JobStep::getTemplateId).distinct()
                .collect(Collectors.toList());
        List<JobStepTemplate> jobStepTemplateList = jobStepTemplateService.list(Wrappers.<JobStepTemplate>lambdaQuery()
                .in(JobStepTemplate::getId, templateIdList));
        HashMap<Integer, JobStepTemplate> jobStepTemplateMap = jobStepTemplateList.stream().collect(
                HashMap::new,
                (map, template) -> map.put(template.getId(), template),
                (map1, map2) -> map1.putAll(map2));

        StringBuilder cmd = new StringBuilder();
        for (JobStep step : jobStepList) {
            String script = jobStepTemplateMap.get(step.getTemplateId()).getScript();
            String templateParameter = step.getTemplateParameter();
            Map<String, String> parameterMap = objectMapper.readValue(templateParameter, Map.class);
            String[] searchList = new String[parameterMap.keySet().size()];
            String[] replacementList = new String[parameterMap.keySet().size()];
            int i = 0;
            for (Map.Entry<String, String> entry : parameterMap.entrySet()) {
                searchList[i] = "{{" + entry.getKey() + "}}";
                replacementList[i] = entry.getValue();
                i++;
            }
            String arg = StringUtils.replaceEachRepeatedly(script, searchList, replacementList);
            log.info(script);
            log.info(templateParameter);
            log.info(arg);
            cmd.append(StringUtils.appendIfMissing(arg, "\n"));
        }
        args.add(StringUtils.removeEnd(cmd.toString(), "\n"));
        return args;
    }

    public static void main(String[] args) {
        System.out.println(StringUtils.appendIfMissing("print('hello')", ";"));
        System.out.println(StringUtils.appendIfMissing("print('world');", ";"));
    }
}
